package com.sh.data.engine;

import com.alibaba.fastjson.JSON;
import com.sh.data.engine.kafka.EngineKafkaConsumer;
import com.sh.data.engine.mbean.Context;
import com.sh.data.engine.mbean.Point;
import com.sh.data.engine.mqtt.EngineMqttConsumer;
import com.sh.data.engine.vo.OnlineTaskParam;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;

import javax.management.MBeanServer;
import javax.management.MBeanServerFactory;
import javax.management.ObjectName;
import javax.management.remote.JMXConnectorServer;
import javax.management.remote.JMXConnectorServerFactory;
import javax.management.remote.JMXServiceURL;
import java.io.File;
import java.io.IOException;
import java.rmi.registry.LocateRegistry;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Data
@Slf4j
public class Engine {

    public static void main(String[] args) {
        Context context = new Context();

        try {

            log.info("start job ...");

            // 在online-task中已知 首参数是json文件的地址
            OnlineTaskParam onlineTaskParam = getJobContent(args[0]);
            //      OnlineTaskParam onlineTaskParam =
            //          getJobContent("/Users/ys/Downloads/5aa3120b-bdce-4fa2-90a5-7132b6b4756c.json");

            context.setType(onlineTaskParam.getType());
            startJmxMonitor(onlineTaskParam.getJmxPort(), context);

            log.info("读取测点名称映射表 ...");
            // 若没有把csv文件的地址带过来, 从默认地址读取文件
            String filePath =
                onlineTaskParam.getRealTimePointMappingCsvFile() == null
                    ? "/data/onlinetask_client/real_time_point_mapping.csv"
                    : onlineTaskParam.getRealTimePointMappingCsvFile();

            Map<String, String> csvData = getCsvData(filePath);

            if ("MQTT".equalsIgnoreCase(onlineTaskParam.getType())) {
                EngineMqttConsumer engineMqttConsumer = new EngineMqttConsumer(onlineTaskParam, context);
                engineMqttConsumer.init();
            } else {
                // 葫芦素项目，需要传入测点名称映射表信息
                EngineKafkaConsumer engineKafkaConsumer =
                    new EngineKafkaConsumer(onlineTaskParam, context, csvData);
                engineKafkaConsumer.init();
                engineKafkaConsumer.start();
            }
        } catch (Throwable e) {
            log.error("\n\n经智能分析,该任务最可能的错误原因是:\n" + ExceptionTracker.trace(e));
            log.info("job end...");

            System.exit(1);
        }

        log.info("job end...");
        System.exit(0);
    }

    private static void startJmxMonitor(int jmxPort, Context context) {
        try {
            LocateRegistry.createRegistry(jmxPort); // 必须要这句，参数为端口号
            String jmxServerName = "client";
            ObjectName helloName = new ObjectName(jmxServerName + ":name=point");
            Point hello = new Point(context);
            MBeanServer server = MBeanServerFactory.createMBeanServer(jmxServerName);

            server.registerMBean(hello, helloName);

            JMXServiceURL url =
                new JMXServiceURL(
                    "service:jmx:rmi:///jndi/rmi://localhost:" + jmxPort + "/" + jmxServerName);
            JMXConnectorServer jmxConnServer =
                JMXConnectorServerFactory.newJMXConnectorServer(url, null, server);
            jmxConnServer.start();

        } catch (Exception e) {
            log.error(e.getMessage(), e);
            throw new RuntimeException("启动监听失败:" + e.getMessage());
        }
    }

    /**
     * 读取的json文件的内容
     */
    private static OnlineTaskParam getJobContent(String jobResource) {

        String json;
        File file = new File(jobResource);
        try {
            json = FileUtils.readFileToString(file, "UTF-8");
        } catch (IOException e) {
            throw new RuntimeException("读取文件失败:" + e.getMessage());
        }

        if (StringUtils.isBlank(json)) {
            throw new RuntimeException("解析文件失败:" + jobResource + "文件为空");
        }

        if (file.exists()) {
            // file.delete();
        }

        return JSON.parseObject(json, OnlineTaskParam.class);
    }

    private static Map<String, String> getCsvData(String filePath) {
        Map<String, String> csvData = new HashMap<>();
        try {
            File file = new File(filePath);
            if (!file.exists()) {
                // 如果文件不存在，直接返回
                log.info("无测点名称映射表，跳过...");
                return csvData;
            }

            List<String> dataList = FileUtils.readLines(new File(filePath), "UTF-8");
            for (String data : dataList) {
                String[] values = data.split(",");
                csvData.put(values[0], values[1]);
            }
        } catch (Exception e) {
            log.error("读取测点名称映射表", e);
        }
        return csvData;
    }
}
